//package com.study.spark.scala.streaming.kafka.zk
//
//import org.apache.commons.codec.StringDecoder
//import org.apache.spark.SparkConf
//import org.apache.spark.streaming.{Seconds, StreamingContext}
//
///**
//  *
//  * @author stephen
//  * @create 2019-03-20 16:06
//  * @since 1.0.0
//  */
//object KafakDirectOffsetUseZK {
//
//  def main(args: Array[String]): Unit = {
//    val topic = Array("test")
//    val groupId = "test_group"
//
//    def initKafkaParams = {
//      Map[String, String](
//        "metadata.broker.list" -> "localhost:9092",
//        "group.id " -> groupId,
//        "fetch.message.max.bytes" -> "20971520",
//        "auto.offset.reset" -> "smallest"
//      )
//    }
//
//    val conf = new SparkConf().setAppName("Kafka Offset Redis").setMaster("local[2]")
//    val ssc = new StreamingContext(conf, Seconds(10))
//
//    // kafka参数
//    val kafkaParams = initKafkaParams
//    val manager = new KafkaManager(kafkaParams)
//    val kafkaStream = manager.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topic.toSet)
//
//    kafkaStream.foreachRDD(rdd=>{
//      rdd.map(_._2)
//      // 更新offsets
//      manager.updateZKOffsets(rdd)
//
//    })
//  }
//}
